package cn.edu.thu.laud.thrift;
import java.nio.ByteBuffer;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.hadoop.hive.service.HiveClient;
import org.apache.hadoop.hive.service.HiveServerException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import cn.edu.thu.laud.jdbc.util.LaUDJdbcUtils;
import cn.edu.thu.laud.server.thrift.LaUD;
import cn.edu.thu.laud.server.thrift.LaUDServerException;



public class LaUDClient extends LaUD.Client{
	static ByteBuffer emptyBuffer=null; 
	static{
		emptyBuffer=LaUDJdbcUtils.compressQuery("",	 Compression.GZIP); 
	}

	Cassandra.Client cassandraClient;
	//HiveInterface hiveClient; //this is a local mode. we can consider that it use a personal server itself.
	HiveClient hiveClient;

	private  String sessionId;

	public LaUDClient(TProtocol iprot, TProtocol oprot) throws Exception {
		super(iprot, oprot);
		cassandraClient=new LaUDCassandraClient(iprot,oprot);
		//hiveClient=new HiveClient(iprot,oprot);
		throw new Exception("Hive require only one prot...Try to use public LaUDClient(TProtocol prot) instead;");
	}
	public LaUDClient(TProtocol prot) throws LaUDServerException, TException {
		super(prot);
		cassandraClient=new LaUDCassandraClient(prot);
		hiveClient=new LaUDHiveClient(prot);

	}
	public void setSessionId(){
		try {
			this.sessionId=this.get_laud_session_id();
		} catch (LaUDServerException e) {
			e.printStackTrace();
		} catch (TException e) {
			e.printStackTrace();
		}
	}
	public Cassandra.Client getCassandraClient() {
		return cassandraClient;
	}
	public HiveClient getHiveClient() {
		return hiveClient;
	}

	public String getSessionId() {
		return sessionId;
	}

	public class LaUDCassandraClient extends Cassandra.Client{

		/**
		 * replace super.execute_cql_query() function's result.
		 */
		CqlResult fakeCqlResult=null;
		public LaUDCassandraClient(TProtocol arg0) {
			super(arg0);
		}

		public LaUDCassandraClient(TProtocol iprot, TProtocol oprot) {
			super(iprot,oprot);
		}

		public CqlResult getFakeCqlResult() {
			return fakeCqlResult;
		}

		public void setFakeCqlResult(CqlResult fakeCqlResult) {
			this.fakeCqlResult = fakeCqlResult;
		}
		/**
		 * before run this function,should run setFakeCqlResult() first.
		 */
		@Override
		public CqlResult execute_cql_query(ByteBuffer query,
				Compression compression) throws InvalidRequestException,
				UnavailableException, TimedOutException,
				SchemaDisagreementException, TException {

			if(query.equals(emptyBuffer))
				return  fakeCqlResult;
			else {
				return super.execute_cql_query(query, compression);
			}
		}
		public CqlResult executeCql(ByteBuffer query,
				Compression compression) throws InvalidRequestException,
				UnavailableException, TimedOutException,
				SchemaDisagreementException, TException {
			return super.execute_cql_query(query, compression);
		}

	}
	class LaUDHiveClient extends HiveClient{

		public LaUDHiveClient(TProtocol prot) {
			super(prot);
		}
		/**
		 * this method do nothing.the super.execute(String) is replaced by executeHiveQL(String)
		 */
		@Override
		public void execute(String query) throws HiveServerException,
		TException {
			if(!query.equals(""))
				executeHiveQl(query);
		}
		public void executeQL(String query) throws HiveServerException,TException{
			executeHiveQl(query);
		}	
	}
	/**
	 * replace the close() function
	 * @author hxd
	 *
	 */
	public static class LaUDNullCloseTTransport extends TTransport{
		TTransport transport;

		public boolean isOpen() {
			return transport.isOpen();
		}

		public void open() throws TTransportException {
			transport.open();
		}

		public int read(byte[] buf, int off, int len)
				throws TTransportException {
			return transport.read(buf, off, len);
		}

		public void write(byte[] buf, int off, int len)
				throws TTransportException {
			transport.write(buf, off, len);
		}

		public void consumeBuffer(int len) {
			transport.consumeBuffer(len);
		}

		public boolean equals(Object obj) {
			return transport.equals(obj);
		}

		public void flush() throws TTransportException {
			transport.flush();
		}

		public byte[] getBuffer() {
			return transport.getBuffer();
		}

		public int getBufferPosition() {
			return transport.getBufferPosition();
		}

		public int getBytesRemainingInBuffer() {
			return transport.getBytesRemainingInBuffer();
		}

		public int hashCode() {
			return transport.hashCode();
		}

		public boolean peek() {
			return transport.peek();
		}

		public int readAll(byte[] buf, int off, int len)
				throws TTransportException {
			return transport.readAll(buf, off, len);
		}

		public String toString() {
			return transport.toString();
		}

		public void write(byte[] buf) throws TTransportException {
			transport.write(buf);
		}

		public LaUDNullCloseTTransport(TTransport transport) {
			super();
			this.transport = transport;
		}


		/**
		 * do nothing
		 */
		@Override
		public void close() {
			;
		}



	}




}
